package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.gmall.realtime.app.func.DimSink;
import com.atguigu.gmall.realtime.app.func.MyDeserializationSchema;
import com.atguigu.gmall.realtime.app.func.TableProcessFunction;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * Author: Felix
 * Date: 2022/3/12
 * Desc: 业务数据动态分流
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、BaseDBApp
 * 分流代码编写思路
     业务数据动态分流
     业务系统产生业务数据保存到MySQL数据库
     binlog记录业务数据库变化数据
     maxwell伪装成从机，从binlog中读取变化的数据
     maxwell会将读取到的变化数据封装为json字符串发送到kafka的ods_base_db_m
     BaseDBApp从ods_base_db_m主题中读取数据
        基本环境准备
        检查点相关设置
        从ods_base_db_m读取数据
        对读取的数据进行类型转换	jsonStr->jsonObj
        对数据进行简单的ETL

        使用FlinkCDC读取配置表中的数据
        将读取到的配置流进行广播并定义广播状态

        将业务流和广播流进行关联 connect
        对关联之后的数据进行处理 process
            processElement-处理业务流数据
                4.字段过滤
                2.根据当前处理业务数据，获取表名以及操作类型，将表名和操作类型拼接为key
                根据key从广播状态中获取对应的配置信息，根据配置信息判断是事实还是维度
                事实---主流
                维度---侧输出流
             processBroatcastElement-处理广播流数据
                3.如果读取到的配置信息是维度并是insert操作，提前创建维度表
                1.读取广播流中一条配置信息，封装为TableProcess对象，放到广播状态中
        将维度侧输出流的数据写到Phoenix表中
        自定义实体类DimSink实现SinkFunction，重写invoke方法
            -拼接upsert语句
            -通过jdbc执行upsert语句  注意：手动提交事务
        将主流事实数据写到kafka不同主题中
            获取FlinkKafkaProducer对象
            new FlinkKafkaProducer(default_topic,序列化,props,一致性)
     在BaseDBApp中调用工具类中获取FlinkKafkaProducer对象的方法，并实现序列化
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置(略)
        //TODO 3.从Kafka中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "ods_base_db_m";
        String groupId = "base_db_app_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaSource);
        //TODO 4.对读取的流中的数据进行类型转换    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);

        //TODO 5.简单的ETL
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.filter(
            new FilterFunction<JSONObject>() {
                @Override
                public boolean filter(JSONObject jsonObj) throws Exception {
                    boolean flag = jsonObj.getString("table") != null &&
                        jsonObj.getString("table").length() > 0 &&
                        jsonObj.getJSONObject("data") != null &&
                        jsonObj.getString("data").length() > 3;
                    return flag;
                }
            }
        );
        //filterDS.print(">>>>");

        //TODO 6.使用FlinkCDC读取配置表中数据
        DebeziumSourceFunction<String> mySqlSourceFunction = MySQLSource.<String>builder()
            .hostname("hadoop202")
            .port(3306)
            .databaseList("gmall0906_realtime")
            .tableList("gmall0906_realtime.table_process")
            .username("root")
            .password("123456")
            .startupOptions(StartupOptions.initial())
            .deserializer(new MyDeserializationSchema())
            .build();

        DataStreamSource<String> mysqlStrDS = env.addSource(mySqlSourceFunction);

        //TODO 7.将读取到的配置表流进行广播 声明广播状态
        //7.1 定义广播状态描述器
        MapStateDescriptor<String, TableProcess> mapStateDescriptor =
            new MapStateDescriptor<String, TableProcess>("mapStateDescriptor",String.class,TableProcess.class);
        //7.2 广播
        BroadcastStream<String> broadcastDS = mysqlStrDS.broadcast(mapStateDescriptor);

        //TODO 8.将主流业务数据和广播流的配置数据进行关联  connect
        BroadcastConnectedStream<JSONObject, String> connectDS = filterDS.connect(broadcastDS);

        //TODO 9.分流     事实--主流      维度数据--维度侧输出流中
        OutputTag<JSONObject> dimTag = new OutputTag<JSONObject>("dimTag") {};
        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(
            new TableProcessFunction(dimTag,mapStateDescriptor)
        );
        DataStream<JSONObject> dimDS = realDS.getSideOutput(dimTag);
        realDS.print(">>>");
        dimDS.print("####");

        //TODO 10.将维度侧输出流数据写到Phoenix表
        dimDS.addSink(new DimSink());

        //TODO 11.将主流数据写到kafka主题中
        realDS.addSink(
            MyKafkaUtil.getKafkaSinkByCustom(
                new KafkaSerializationSchema<JSONObject>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, @Nullable Long timestamp) {
                        String topic = jsonObj.getString("sink_table");
                        return new ProducerRecord<byte[], byte[]>(topic,jsonObj.getJSONObject("data").toJSONString().getBytes());
                    }
                }
            )
        );

        env.execute();

    }
}
